package com.yjzx.util.binlog.core.handle;

import com.yjzx.util.binlog.core.config.BinLogBean;
import com.yjzx.util.binlog.core.constants.MessageConstant;
import com.yjzx.util.binlog.core.executor.BinLogExecutor;
import com.yjzx.util.binlog.core.listener.MysqlBinLogListener;
import com.yjzx.util.binlog.core.util.BinLogUtils;
import lombok.extern.slf4j.Slf4j;

import java.util.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

/**
 * @author yjzx
 * @date 2024/9/20
 * @description 初始化binlog监听器的处理器
 */
@Slf4j
public class BinlogHandle {
    private final BinLogBean binLogConstants;
    private final MysqlBinLogListener mysqlBinLogListener;
    private final Set<BinLogExecutor> binLogExecutors = new HashSet<>();
    private volatile boolean flag = true;


    public BinlogHandle(BinLogBean binLogConstants, MysqlBinLogListener mysqlBinLogListener, List<BinLogExecutor> binLogListeners) {
        this.binLogConstants = binLogConstants;
        this.mysqlBinLogListener = mysqlBinLogListener;
        this.binLogExecutors.addAll(binLogListeners);
    }

    /**
     * 添加执行器
     *
     * @param binLogExecutor
     * @return
     */
    public boolean addBinLogExecutor(BinLogExecutor... binLogExecutor) {
        return this.binLogExecutors.addAll(Arrays.asList(binLogExecutor));
    }

    /**
     * build方式添加执行器
     *
     * @param binLogExecutor
     * @return
     */
    public BinlogHandle thenAddBinLogExecutor(BinLogExecutor... binLogExecutor) {
        addBinLogExecutor(binLogExecutor);
        return this;
    }

    /**
     * 启动监听器
     */
    public void start() {
        if (flag) {
            flag = false;
        } else {
            return;
        }
        init();
        // 初始化监听器
        mysqlBinLogListener.finish();
    }

    /**
     * 初始化并处理监听器
     */
    private void init() {
        Map<String, AtomicInteger> tableNumMap = new HashMap<>();
        log.info(MessageConstant.BINLOG_REGISTER_LISTENER, binLogExecutors.size());
        for (Map.Entry<String, String> db : binLogConstants.getDbTable().entrySet()) {
            String dbName = db.getKey();

            List<String> tableList = BinLogUtils.getListByStr(db.getValue());
            try {
                Set<String> tables = tableList.stream().map(x -> BinLogUtils.getdbTable(dbName, x)).collect(Collectors.toSet());
                // 注册监听
                binLogExecutors.forEach(binLogListener ->
                        tableRegNum(tableNumMap, mysqlBinLogListener.regListeners(dbName, tables, binLogListener)));
            } catch (
                    Exception e) {
                log.error(MessageConstant.BINLOG_REGISTER_LISTENER_ERROR, e);
            }
        }
        tableNumMap.forEach((key, value) -> log.info(MessageConstant.BINLOG_REGISTER_LISTENER_SUCCESS, key, value.get()));
    }

    /**
     * 监听数量
     *
     * @param map
     * @param strings
     */
    private void tableRegNum(Map<String, AtomicInteger> map, List<String> strings) {
        strings.forEach(x -> {
            map.putIfAbsent(x, new AtomicInteger(0));
            map.get(x).incrementAndGet();
        });
    }
}
